package com.example.tutorialkafka.consumer;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.sql.Timestamp;

/**
 * @author: cph
 * @date: 2021/11/11 16:17
 */
@Slf4j
@Component
public class KafkaConsumer {

    @KafkaListener(topics = "topic_growth_online")
    public void receiveMessage(ConsumerRecord<String, String> record) {
        try {
            log.info("online kafka record:[{}]", record);
            JSONObject recordValue = JSONObject.parseObject(record.value());
            String eventType = recordValue.get("EventType").toString();
            JSONArray rows = (JSONArray) recordValue.get("Rows");
            if ("insert".equals(eventType) || "update".equals(eventType)) {
                log.info("online kafka insert&update record:[{}]", record);
                JSONObject obj = (JSONObject) rows.get(rows.size() - 1);
                // String id = obj.get("id").toString();
                String userId = obj.get("user_id").toString();
                String tuiguangId = obj.get("tuiguang_id").toString();
                String generalizeId = obj.get("generalize_id").toString();
                String url = obj.get("url").toString();
                Timestamp updateTime = Timestamp.valueOf(obj.get("update_time").toString());

                // 业务逻辑
                //mongoDBService.upsert(userId, tuiguangId, generalizeId, url, updateTime);
            }
        } catch (Exception e) {
            log.error("KafkaConsumer数据处理异常", e);
        }
    }
}
